package com.wang.sql.util

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import org.json4s.{Formats, NoTypeHints}
import java.lang.Thread.sleep
import java.util.Properties
import org.json4s.jackson.Serialization
import org.json4s.jackson.Serialization.write
/**
 * @Desc kafka 工具类
 * @Author wang bo
 * @Date 2021/5/17 14:28
 */
object KafkaUtil {
  def main(args: Array[String]): Unit = {
    kafka_produce
  }

  def init_kafka(): KafkaProducer[String,String] ={
    val properties:Properties = new Properties()
    properties.put("bootstrap.servers", "devcdh1:9092,devcdh2:9092,devcdh3:9092")
    properties.put("group.id", "bigdata")
    properties.put("acks", "all")
    properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    new KafkaProducer(properties)
  }

  def kafka_produce(): Unit ={
    val producer: KafkaProducer[String,String] = init_kafka()
    implicit val formats: AnyRef with Formats = Serialization.formats(NoTypeHints)

    // 测试滚动窗口
//    test_tumble_window
   // 测试 over窗口
    test_over_window
    // 测试窗口聚合clue
//    test_clue
      .foreach(item=>{
      val message=write(item)
      println(message)
      producer.send(new ProducerRecord[String,String]("testtopic",message))
      sleep(3000)
    })

    producer.close()
  }

  def test_tumble_window(): Array[AnyRef] ={
    Array(
      Map(
        "user_name"->"xiao ming",
        "click_url"->"www.baidu.com",
        "update_time"->1621232023000L
      ),
      Map(
        "user_name"->"xiao ming",
        "click_url"->"www.baidu.com",
        "update_time"->1621232024000L
      ),
      Map(
        "user_name"->"xiao ming",
        "click_url"->"www.baidu.com",
        "update_time"->1621232025000L
      ),
      Map(
        "user_name"->"xiao ming",
        "click_url"->"www.baidu.com",
        "update_time"->1621232026000L
      ),
      Map(
        "user_name"->"xiao ming",
        "click_url"->"www.baidu.com",
        "update_time"->1621232027000L
      ),
      Map(
        "user_name"->"xiao ming",
        "click_url"->"www.baidu.com",
        "update_time"->1621232029000L
      ),
      Map(
        "user_name"->"xiao ming",
        "click_url"->"www.baidu.com",
        "update_time"->1621232030000L
      ),
      Map(
        "user_name"->"xiao ming",
        "click_url"->"www.baidu.com",
        "update_time"->1621232031000L
      ),
      Map(
        "user_name"->"xiao ming",
        "click_url"->"www.baidu.com",
        "update_time"->1621232032000L
      ),
      Map(
        "user_name"->"xiao ming",
        "click_url"->"www.baidu.com",
        "update_time"->1621232033000L
      )
    )
  }




  def test_tumble_window2(): Array[AnyRef] ={
    Array(
      Map(
        "user_name"->"xiao ming",
        "click_url"->"www.baidu.com",
        "update_time"->"2020-04-15 08:05"
      ),
      Map(
        "user_name"->"xiao ming",
        "click_url"->"www.baidu.com",
        "update_time"->"2020-04-15 08:07"
      ),
      Map(
        "user_name"->"xiao ming",
        "click_url"->"www.baidu.com",
        "update_time"->"2020-04-15 08:09"
      ),
      Map(
        "user_name"->"xiao ming",
        "click_url"->"www.baidu.com",
        "update_time"->"2020-04-15 08:11"
      ),
      Map(
        "user_name"->"xiao ming",
        "click_url"->"www.baidu.com",
        "update_time"->"2020-04-15 08:13"
      ),
      Map(
        "user_name"->"xiao ming",
        "click_url"->"www.baidu.com",
        "update_time"->"2020-04-15 08:15"
      ),
      Map(
        "user_name"->"xiao ming",
        "click_url"->"www.baidu.com",
        "update_time"->"2020-04-15 08:17"
      )
    )

  }


  def test_over_window(): Array[AnyRef] ={
    Array(
      Map(
        "item_id"->"ITEM001",
        "item_type"->"apple",
        "event_time"->"2020-04-15 10:01:00",
        "price"->20
      ),
      Map(
        "item_id"->"ITEM002",
        "item_type"->"apple",
        "event_time"->"2020-04-15 10:02:00",
        "price"->50
      ),
      Map(
        "item_id"->"ITEM003",
        "item_type"->"apple",
        "event_time"->"2020-04-15 10:03:00",
        "price"->30
      ),
      Map(
        "item_id"->"ITEM004",
        "item_type"->"apple",
        "event_time"->"2020-04-15 10:04:00",
        "price"->60
      ),
      Map(
        "item_id"->"ITEM005",
        "item_type"->"apple",
        "event_time"->"2020-04-15 10:05:00",
        "price"->40
      ),
      Map(
        "item_id"->"ITEM006",
        "item_type"->"apple",
        "event_time"->"2020-04-15 10:06:00",
        "price"->20
      ),
      Map(
        "item_id"->"ITEM007",
        "item_type"->"apple",
        "event_time"->"2020-04-15 10:07:00",
        "price"->10
      ),
      Map(
        "item_id"->"ITEM008",
        "item_type"->"orange",
        "event_time"->"2020-04-15 10:08:00",
        "price"->20
      ),
      Map(
        "item_id"->"ITEM009",
        "item_type"->"orange",
        "event_time"->"2020-04-15 10:09:00",
        "price"->21
      ),
      Map(
        "item_id"->"ITEM010",
        "item_type"->"apple",
        "event_time"->"2020-04-15 10:10:00",
        "price"->22
      ),
      Map(
        "item_id"->"ITEM011",
        "item_type"->"orange",
        "event_time"->"2020-04-15 10:11:00",
        "price"->26
      )
    )
  }


  def test_clue(): Array[AnyRef] ={
    Array(
      Map(
        "item_id"->"ITEM001",
        "item_type"->"apple",
        "event_time"->"2020-04-15 10:01:00",
        "price"->20
      ),
      Map(
        "item_id"->"ITEM001",
        "item_type"->"apple",
        "event_time"->"2020-04-15 10:02:00",
        "price"->50
      ),
      Map(
        "item_id"->"ITEM001",
        "item_type"->"apple",
        "event_time"->"2020-04-15 10:03:00",
        "price"->30
      ),
      Map(
        "item_id"->"ITEM001",
        "item_type"->"apple",
        "event_time"->"2020-04-15 10:04:00",
        "price"->60
      ),
      Map(
        "item_id"->"ITEM001",
        "item_type"->"apple",
        "event_time"->"2020-04-15 10:05:00",
        "price"->40
      ),
      Map(
        "item_id"->"ITEM002",
        "item_type"->"apple",
        "event_time"->"2020-04-15 10:06:00",
        "price"->20
      ),
      Map(
        "item_id"->"ITEM002",
        "item_type"->"apple",
        "event_time"->"2020-04-15 10:07:00",
        "price"->10
      ),
      Map(
        "item_id"->"ITEM002",
        "item_type"->"orange",
        "event_time"->"2020-04-15 10:08:00",
        "price"->20
      ),
      Map(
        "item_id"->"ITEM002",
        "item_type"->"orange",
        "event_time"->"2020-04-15 10:09:00",
        "price"->21
      ),
      Map(
        "item_id"->"ITEM02",
        "item_type"->"apple",
        "event_time"->"2020-04-15 10:10:00",
        "price"->22
      ),
      Map(
        "item_id"->"ITEM02",
        "item_type"->"orange",
        "event_time"->"2020-04-15 10:11:00",
        "price"->26
      )
    )
  }

}
